package com.aispeech.asr.apis.detection.websocket.handler;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.stereotype.Component;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.cors.reactive.CorsConfigurationSource;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.server.ServerWebExchange;

import com.aispeech.asr.apis.detection.callback.SedOnlineCallback;
import com.aispeech.asr.apis.detection.config.AppConfig;
import com.aispeech.asr.apis.detection.config.DopcConfig;
import com.aispeech.asr.apis.detection.constants.enums.ErrorDescEnum;
import com.aispeech.asr.apis.detection.exceptions.BusinessException;
import com.aispeech.asr.apis.detection.websocket.connection.EmitterManager;
import com.aispeech.asr.apis.detection.websocket.emitter.EmitterState;
import com.aispeech.asr.apis.detection.websocket.emitter.EmitterType;
import com.aispeech.asr.comm.data.vo.ResultOnline;
import com.aispeech.jarpc.JArpc;
import com.aispeech.jarpc.op.sed.SedRequest;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Slf4j
@Component
public class AsrWsHandler implements WebSocketHandler, CorsConfigurationSource {
    protected final static Charset CHARSET_UTF8 = StandardCharsets.UTF_8;

    @Autowired
    private EmitterManager manager;
    
    @Autowired
    private AppConfig config;
    
    @Autowired
    private DopcConfig DOpsConfig;

    public EmitterManager getManager() {
        return manager;
    }


    @Override
    public Mono<Void> handle(WebSocketSession session) {
    	
    	String clientSessionId = session.getId();
    	log.info("Receive from client. session id: {},HandshakeInfo:{},manager:{}",clientSessionId,session.getHandshakeInfo(),manager);
    	//addPrometheusMetrics();
    	
    	Mono<Void> in = session.receive()
                 .doOnNext(message -> received(session, message))
                 .doOnError(throwable -> {
                	 log.info("Connected from client close. session id: {}",clientSessionId);  
                     manager.removeClientEmitter(clientSessionId);
                 }).then();
    	Mono<Void> out = manager.addSession(session);	
        
        return Mono.zip(in, out).doAfterTerminate(()->{
        	log.info("zip close. session id: {} ",clientSessionId); 
        	manager.backupAudio(clientSessionId); //正常关闭，判断是否存过
        	manager.removeClientEmitter(clientSessionId);
        	//decPrometheusMetrics();
        }).doOnError(throwable ->{
        	log.info("zip execute doOnError,session id: {}",clientSessionId, throwable);
        	manager.backupAudio(clientSessionId); //客户端异常关闭
        	manager.removeClientEmitter(clientSessionId);
        	//decPrometheusMetrics();
        }).then();

    }
    
    @SneakyThrows
    private void received(WebSocketSession session, WebSocketMessage message) {
    	
    	String result = "";
    	
    	manager.addTimeToClientReceivedList(session.getId());
        WebSocketMessage.Type type = message.getType();
        //i++;
        switch (type) {
            case TEXT:
            	parsePayloadAsText(session,message);
                break;
            case BINARY:
            	int length = message.getPayload().readableByteCount();
            	if(!getManager().isClientRunning(session.getId())) {
            		manager.immediateSendCustomerClient(session.getId(), JSON.toJSONString(ResultOnline.fail(ErrorDescEnum.ERROR_21003.getCode(),ErrorDescEnum.ERROR_21003.getMessage())));
            		log.error("Receive from client. client state is error。 session id: {},clientState:{},length:{}",session.getId(),getManager().getClientState(session.getId()),length);
            		//throw new BusinessException(ErrorDescEnum.ERROR_21003.getCode(),ErrorDescEnum.ERROR_21003.getMessage());
            		//result =  JSON.toJSONString(Result.fail(ErrorDescEnum.ERROR_21003.getCode(),ErrorDescEnum.ERROR_21003.getMessage()));
            	}
            	if(length == 0) {
            		manager.updateClientState(session.getId(), EmitterState.RECEIVE_STOP);
            		log.info("Receive from client. received stop info 。 session id: {}, time:{}",session.getId(),System.currentTimeMillis());
            	}
            	manager.doSed(session.getId(), message.getPayload());
            	//onBinaryMessageReceived(session, message.getPayload());
                break;
            case PING:
            	log.info("[ received ] received PING message. type: {}, text: {}", type, message.getPayloadAsText());
                break;
            case PONG:
            	log.info("[ received ] received PONG message. type: {}, text: {}", type, message.getPayloadAsText());
                break;
            default:
                log.info("[ received ] received other message. type: {}, text: {}", type, message.getPayloadAsText());
                break;
        }
        
        //return result;
    }
    
    private String onTextMessageReceived(WebSocketSession session,  String asrMessage) {
        
    	String result = "";
    	
    	result = "[ onTextMessageReceived ] received a Text message:" + asrMessage + ", length: " + asrMessage.length();
        
    	result = JSON.toJSONString(ResultOnline.ok(result));
    	
        return result;
    }

    private String onBinaryMessageReceived(WebSocketSession session, DataBuffer dataBuffer) {
        
    	String result = "";
    	
    	result = "[ onBinaryMessageReceived ] received a Binary message, length: " + dataBuffer.readableByteCount();
    	
    	result = JSON.toJSONString(ResultOnline.ok(result));
        
        return result;
    }
    
    private WebSocketMessage response(WebSocketSession session, String result) {
        WebSocketMessage message = null;
        
        
        if (StringUtils.isNotEmpty(result)) {
            message = session.textMessage(result);
        } else {
            log.warn("no data need to send to: {}.", session.getHandshakeInfo().getRemoteAddress());
        }
        
        return message;
    }
    
    
    @SneakyThrows
    private void parsePayloadAsText(WebSocketSession session,WebSocketMessage message) {
    	String clientSessionId = session.getId();
    	
    	if(getManager().getClientState(clientSessionId) != EmitterState.READY) {
    		manager.immediateSendCustomerClient(clientSessionId, JSON.toJSONString(ResultOnline.fail(ErrorDescEnum.ERROR_21003.getCode(),ErrorDescEnum.ERROR_21003.getMessage())));
    		log.error("Receive from client. client state is error。 session id: {},clientState:{}",clientSessionId,getManager().getClientState(session.getId()));
    		return;
    	}
    	
    	String payloadAsText = message.getPayloadAsText(CHARSET_UTF8);
    	log.info("Receive from client. session id: {},payloadAsText:{}",clientSessionId,payloadAsText);
    	
    	String requestQuery = session.getHandshakeInfo().getUri().getQuery();
        log.info("Connected from client. clientSessionId:{},requestQuery: {}",clientSessionId, requestQuery);
        
        Map<String,String> parmMap = parseParamsByURI(session);
    	String requestId = UUID.randomUUID().toString().replaceAll("-", "");
    	String productId = "";
    	if(parmMap != null && parmMap.containsKey("requestId")) {
    		requestId = parmMap.get("requestId");
    		productId = parmMap.get("productId");
    	}
    	
    	if(getManager().isClientRunning(clientSessionId)) {
    		manager.immediateSendCustomerClient(clientSessionId, JSON.toJSONString(ResultOnline.fail(ErrorDescEnum.ERROR_21000.getCode(),ErrorDescEnum.ERROR_21000.getMessage())));
    		log.error("[ ======received text info ======] payloadAsText:{}", payloadAsText);
    		return;
    	}
    	
    	JSONObject requestObj = null;
    	JSONObject audioObject = null;
    	String cfgRes = "";
    	try {
    		requestObj = JSON.parseObject(payloadAsText);
    		//校验
    		if(requestObj.containsKey("audio")) {
    			audioObject = requestObj.getJSONObject("audio");
    			
    			Integer sampleRate = audioObject.getInteger("sample_rate") ;
    	    	Integer channel = audioObject.getInteger("channel");
    	    	Integer sampleBytes = audioObject.getInteger("sample_bytes");
    	    	String audioType = audioObject.getString("audio_type");
    			if(sampleRate == null || sampleRate.intValue() != 16000 
    					|| channel == null || channel.intValue() != 1
    					|| sampleBytes == null || sampleBytes.intValue() != 2
    					|| audioType == null || !"wav".equalsIgnoreCase(audioType)
    			) {
    				manager.immediateSendCustomerClient(clientSessionId, JSON.toJSONString(ResultOnline.fail(ErrorDescEnum.ERROR_21002.getCode(),ErrorDescEnum.ERROR_21002.getMessage())));
            		log.error("[ ======received text info with audio error ======] payloadAsText:{}", payloadAsText);
            		return;
    			}
    		}else {
    			manager.immediateSendCustomerClient(clientSessionId, JSON.toJSONString(ResultOnline.fail(ErrorDescEnum.ERROR_21002.getCode(),ErrorDescEnum.ERROR_21002.getMessage())));
        		log.error("[ ======received text info ======] payloadAsText:{}", payloadAsText);
        		return;
    		}
    		
    		if(requestObj.containsKey("request_id") && StringUtils.isNotBlank(requestObj.getString("request_id")) ) {
    			requestId = requestObj.getString("request_id");
    		}
    		
    		String modeType = "default";
    		if(requestObj.containsKey("mode_type") && StringUtils.isNotBlank(requestObj.getString("mode_type")) ) {
    			modeType = requestObj.getString("mode_type");
    		}
    		
            if(StringUtils.isBlank(modeType)) {
            	cfgRes = config.getCfgRelationshipMap().get("default");
            }else {
            	cfgRes = config.getCfgRelationshipMap().get(modeType);
                if(StringUtils.isBlank(cfgRes)) {
                	cfgRes = config.getCfgRelationshipMap().get("default");
                }
            }
            log.info("[ ======received text info ======] session id: {},cfgRes:{}",clientSessionId,config.getResRoot()+cfgRes);
    	}catch (Exception e) {
    		log.error("ConnectServer error.", e);
    		manager.immediateSendCustomerClient(clientSessionId, JSON.toJSONString(ResultOnline.fail(ErrorDescEnum.ERROR_21001.getCode(),ErrorDescEnum.ERROR_21001.getMessage())));
        	log.error("Receive from client. session id: {},message:{}",clientSessionId,message);
        	return;
    	}
    	
    	SedOnlineCallback sedOnlineCallback = new SedOnlineCallback(clientSessionId,manager);
    	JArpc jarpc = new JArpc(DOpsConfig.getHost(), DOpsConfig.getPort(), 60, sedOnlineCallback);
    	
    	jarpc.start(requestId, new SedRequest(config.getResRoot()+cfgRes));
    	
    	getManager().setClientProperties(clientSessionId, requestQuery ,requestId, requestObj, audioObject, jarpc, false);
    	getManager().updateClientState(clientSessionId, EmitterState.RUNNING);
    	getManager().immediateSendCustomerClient(clientSessionId, JSON.toJSONString(ResultOnline.ok(ErrorDescEnum.start.getMessage())));
    }
    

    /**
     * 解析parm
     * @param session
     * @return
     */
    private Map<String, String> parseParamsByURI(WebSocketSession session){
        Map<String, String> map = new HashMap<>();
        String[] params = Optional.ofNullable(session.getHandshakeInfo().getUri().getQuery()).orElse("").split("&");
        for (String param : params) {
            String[] temp = param.split("=");
            if(temp.length == 2){
                map.put(temp[0],temp[1]);
            }
        }
        return map;
    }
    
//    private void addPrometheusMetrics() {
//    	Prometheus.COUNTER_CLOUDVAD_REQUESTS.labels(Prometheus.Env(), Prometheus.NodeName()).inc();
//    	Prometheus.GAUGE_CLOUDVAD_CONNECTED_CONCURRENCY.labels(Prometheus.Env(), Prometheus.NodeName()).inc();
//    }
//    
//    private void decPrometheusMetrics() {
//    	Prometheus.GAUGE_CLOUDVAD_CONNECTED_CONCURRENCY.labels(Prometheus.Env(), Prometheus.NodeName()).dec();;
//    }
    

    @Override
    public CorsConfiguration getCorsConfiguration(ServerWebExchange serverWebExchange) {
        CorsConfiguration configuration = new CorsConfiguration();
        configuration.addAllowedOrigin("*");
        return configuration;
    }
}
